/**
 * @file ttds_cjobq.c
 * @author liangyaozhan (262666882@qq.com)
 * @brief 
 * @version 0.1
 * @date 2022-09-03
 * 
 * @copyright Copyright (c) 2022
 * 
 */

#include <string.h>
#include <stdio.h>
#include "ttds/ttds_cjobq.h"


#if defined(WIN32) || defined(_WIN32) || defined(__CYGWIN__) || defined(MINGW64)|| defined(MINGW32)
#include <windows.h>
static void *__init_win32_critical_section()
{
    CRITICAL_SECTION *ptr = (CRITICAL_SECTION*)malloc(sizeof(CRITICAL_SECTION));
    if (ptr){
        InitializeCriticalSection(ptr);
    }
    return ptr;
}
static void __deinit_win32_critical_section(void *p)
{
    CRITICAL_SECTION *ptr = (CRITICAL_SECTION*)p;
    if (ptr){
        DeleteCriticalSection(ptr);
        free(ptr);
    }
}

#define ttds_lock_create_mutex_ptr() __init_win32_critical_section()
#define ttds_lock_delete_mutex_ptr(ptr) __deinit_win32_critical_section(ptr)
#define ttds_lock_decl(lk) CRITICAL_SECTION *lk = (CRITICAL_SECTION*)p_this->p_lock
#define ttds_lock(lk)   EnterCriticalSection(lk)
#define ttds_unlock(lk) LeaveCriticalSection(lk)
#define my_malloc malloc
#define my_free free

#elif defined(RT_THREAD)
#include "rtthread.h"
#define ttds_lock_create_mutex_ptr() NULL
#define ttds_lock_delete_mutex_ptr(ptr) do{}while(0)
#define ttds_lock_decl(lk) rt_base_t lk
#define ttds_lock(lk)   lk = rt_hw_interrupt_disable()
#define ttds_unlock(lk) rt_hw_interrupt_enable(lk)
#define my_malloc rt_malloc
#define my_free rt_free
#else

#ifndef ttds_lock_create_mutex_ptr
#define ttds_lock_create_mutex_ptr() NULL
#endif
#ifndef ttds_lock_delete_mutex_ptr
#define ttds_lock_delete_mutex_ptr(ptr) 
#endif
#ifndef ttds_lock_decl
#define ttds_lock_decl(lk) ttds_hw_interrupt_decl(lk)
#endif
#ifndef ttds_lock
#define ttds_lock(lk)   ttds_hw_interrupt_disable(lk)
#endif
#ifndef ttds_unlock
#define ttds_unlock(lk) ttds_hw_interrupt_enable(lk)
#endif
#ifndef my_malloc
#define my_malloc(s) NULL
#define my_free(p) do{}while (0)
#endif

#endif
#undef ttds_max
#undef ttds_min
#define ttds_max(a,b) (((a) > (b)) ? (a) : (b))
#define ttds_min(a,b) (((a) < (b)) ? (a) : (b))

void ttds_cjob_init(struct ttds_cjob*p_this)
{
    memset(p_this, 0, sizeof(*p_this));
	ttds_queue_init(&p_this->node);
    p_this->ref = 1;
}
void ttds_cjob_destroy(struct ttds_cjob*p_this)
{
    if (p_this->destroy){
        p_this->destroy(p_this);
    }
}
void ttds_cjob_transaction_init(struct ttds_cjob_transaction *p_this)
{
	ttds_queue_init(&p_this->list_suspended);
	ttds_queue_init(&p_this->node_in_jq);
    p_this->in_running_q_count = 0;
}
void ttds_cjobq_init(struct ttds_cjobq *p_this)
{
	ttds_queue_init(&p_this->job_free);
	ttds_queue_init(&p_this->job_list);
	ttds_queue_init(&p_this->job_list_delayed);
    p_this->p_running_j = 0;
    p_this->tick_next_timeout = 0;
    p_this->tick_pass = 0;
    p_this->p_lock = ttds_lock_create_mutex_ptr();
}

void ttds_cjobq_destroy( struct ttds_cjobq *p_this )
{
    struct ttds_cjob *j = 0;
    struct ttds_cjob *tmp_next = 0;
    ttds_queue_t h;
    ttds_lock_decl(lk);

    ttds_lock(lk);
    ttds_queue_move(&p_this->job_list, &h);
    ttds_unlock(lk);
    ttds_queue_for_each_container_safe(j, tmp_next,&h, struct ttds_cjob,node)
    {
        ttds_cjob_destroy(j);
        ttds_queue_remove(&j->node);
        my_free(j);
    }
    ttds_lock(lk);
    ttds_queue_move(&p_this->job_free, &h);
    ttds_unlock(lk);
    ttds_queue_for_each_container_safe(j, tmp_next,&h, struct ttds_cjob,node)
    {
        ttds_cjob_destroy(j);
        ttds_queue_remove(&j->node);
        my_free(j);
    }
    ttds_lock(lk);
    ttds_queue_move(&p_this->job_list_delayed, &h);
    ttds_unlock(lk);
    ttds_queue_for_each_container_safe(j, tmp_next,&h, struct ttds_cjob,node)
    {
        ttds_cjob_destroy(j);
        ttds_queue_remove(&j->node);
        my_free(j);
    }
    ttds_lock_delete_mutex_ptr(p_this->p_lock);
}

#ifdef TTDS_USING_JOB_HOOK
void ttds_cjobq_set_newjob_hook( struct ttds_cjobq *p_this, void (*f)(void*), void *arg)
{
    p_this->hook = f;
    p_this->arg = arg;
}
#endif
int g_allocated_count = 0;

#ifdef TTDS_STATIC_ALLOC_COUNT

static struct ttds_cjob* ttds_cjobq_list_node_alloc(struct ttds_cjobq *p_this)
{
    static uint8_t inited;
	ttds_queue_t *p = 0;
    ttds_queue_t *h = &p_this->job_free;
    ttds_lock_decl(lk);

    if (!inited){
        static ttds_cjob_t g_cjobs[TTDS_STATIC_ALLOC_COUNT];
        int i;
        inited = 1;
        ttds_lock(lk);
        for (i=0; i<TTDS_STATIC_ALLOC_COUNT; i++){
	        ttds_queue_init( &g_cjobs[i].node );
            ttds_queue_insert_tail( h, &g_cjobs[i].node );
        }
        ttds_unlock(lk);
    }
    ttds_lock(lk);
    if (!ttds_queue_empty(h)){
		p = h->next;
		ttds_queue_remove(p);
		struct ttds_cjob* p_ptr = ttds_queue_container(p, struct ttds_cjob, node);
        ttds_unlock(lk);
		return p_ptr;
    }
    ttds_unlock(lk);
    return NULL;
}

#else
static struct ttds_cjob* ttds_cjobq_list_node_alloc(struct ttds_cjobq *p_this)
{
	ttds_queue_t *p = 0;
    ttds_queue_t *h = &p_this->job_free;
    ttds_lock_decl(lk);
    ttds_lock(lk);
	if (ttds_queue_empty (h))
	{
        ttds_unlock(lk);
		struct ttds_cjob *ptr = (struct ttds_cjob*)my_malloc(sizeof(struct ttds_cjob));
		if (!ptr)
		{
			return 0;
		}
		p = &ptr->node;
        g_allocated_count++;
	    ttds_queue_init(p);
		return ptr;
	} else
	{
		p = h->next;
		ttds_queue_remove(p);
		struct ttds_cjob* p_ptr = ttds_queue_container(p, struct ttds_cjob, node);
        ttds_unlock(lk);
		return p_ptr;
	}
}
void ttds_cjobq_job_pre_allocate(struct ttds_cjobq *p_this, int pre_allocate_count)
{
    int i;
    ttds_lock_decl(lk);

    for (i=0; i<pre_allocate_count; i++ )
    {
        struct ttds_cjob *p = ttds_cjobq_list_node_alloc(p_this);
        ttds_lock(lk);
        ttds_queue_insert_tail( &p_this->job_free, &p->node );
        ttds_unlock(lk);
    }
}
#endif

void list_node_free( ttds_queue_t *h, struct ttds_cjob *p)
{
	ttds_queue_remove(&p->node);
	ttds_queue_insert_tail(h, &p->node );
}

static void __jobq_add_job(struct ttds_cjobq *p_this, struct ttds_cjob *j)
{
    if (j->p_master)
        j->p_master->in_running_q_count++;
    if (j->tick <= 0){
        ttds_queue_insert_tail( &p_this->job_list, &j->node );
    } else {
        if (p_this->tick_next_timeout < 0 || j->tick < p_this->tick_next_timeout){
            p_this->tick_next_timeout = j->tick;
        }
        ttds_queue_insert_tail( &p_this->job_list_delayed, &j->node );
    }
}

void ttds_cjobq_add(struct ttds_cjobq *p_this, void (*f)(struct ttds_cjob*), void*args, void (*d)(struct ttds_cjob*) )
{
    ttds_cjobq_add_delay( p_this, f, args, d, 0 );
}
void ttds_cjobq_add_delay(struct ttds_cjobq *p_this, void (*f)(struct ttds_cjob*), void*args, void (*d)(struct ttds_cjob*), int delay )
{
    struct ttds_cjob *p = ttds_cjobq_list_node_alloc(p_this);
    ttds_lock_decl(lk);
    ttds_cjob_init(p);
    p->func = f;
    p->args = args;
    p->destroy= d;
    p->indexing = 0;
    p->p_master = 0;
    p->p_running_q = 0;
    p->tick = delay;
    ttds_lock(lk);
    __jobq_add_job(p_this, p);
    ttds_unlock(lk);
#ifdef TTDS_USING_JOB_HOOK
    if (p_this->hook){
        p_this->hook( p_this->arg );
    }
#endif
}
void ttds_cjobq_add_seq_begin(struct ttds_cjobq *p_this,struct ttds_cjob_transaction *origin, void (*f)(struct ttds_cjob*), void*args, void (*d)(struct ttds_cjob*)  )
{
    ttds_cjobq_add_seq_begin_delay(p_this, origin, f, args, d, 0 );
}
void ttds_cjobq_add_seq_begin_delay(struct ttds_cjobq *p_this,struct ttds_cjob_transaction *origin, void (*f)(struct ttds_cjob*), void*args, void (*d)(struct ttds_cjob*), int delay )
{
    struct ttds_cjob *p = ttds_cjobq_list_node_alloc(p_this);
    ttds_lock_decl(lk);
    ttds_cjob_init(p);
    p->func = f;
    p->args = args;
    p->destroy= d;
    p->indexing = 0;
    p->p_master = origin;
    p->p_running_q = 0;
    p->tick = delay;
    ttds_assert(p->p_master);
    ttds_lock(lk);
    if ( p->p_master->in_running_q_count>0 )
        ttds_queue_insert_tail( &p->p_master->list_suspended, &p->node );
    else {
        __jobq_add_job(p_this, p);
    }
    ttds_unlock(lk);
#ifdef TTDS_USING_JOB_HOOK
    if (p_this->hook){
        p_this->hook( p_this->arg );
    }
#endif
}

void ttds_cjob_add_delay(const struct ttds_cjob* p_this, void (*f)(struct ttds_cjob*), void* args, void (*d)(struct ttds_cjob*), int delay)
{
    ttds_cjobq_add_delay(p_this->p_running_q, f, args, d, delay);
}
void ttds_cjob_add_seq(const struct ttds_cjob *p_this, void (*f)(struct ttds_cjob*), void*args, void (*d)(struct ttds_cjob*) )
{
    ttds_cjob_add_seq_delay( p_this, f, args, d, 0);
}
void ttds_cjob_add_seq_delay(const struct ttds_cjob *p_this, void (*f)(struct ttds_cjob*), void*args, void (*d)(struct ttds_cjob*), int delay )
{
    ttds_cjobq_add_seq_delay(p_this->p_running_q, p_this, f, args, d, delay);
}
void ttds_cjobq_add_seq(struct ttds_cjobq *p_this,const struct ttds_cjob *parent, void (*f)(struct ttds_cjob*), void*args, void (*d)(struct ttds_cjob*))
{
    ttds_cjobq_add_seq_delay(p_this,parent, f, args, d, 0);
}
void ttds_cjobq_add_seq_delay(struct ttds_cjobq *p_this,const struct ttds_cjob *parent, void (*f)(struct ttds_cjob*), void*args, void (*d)(struct ttds_cjob*), int delay )
{
    struct ttds_cjob *p = ttds_cjobq_list_node_alloc(p_this);
    ttds_lock_decl(lk);
    ttds_cjob_init(p);
    p->func = f;
    p->args = args;
    p->destroy= d;
    p->indexing = parent->indexing+1;
    p->p_master = parent->p_master;
    p->p_running_q = 0;
    p->tick = delay;
    ttds_assert(p->p_master);
    ttds_lock(lk);
    if (p->indexing > 0){
        __jobq_add_job(p_this, p);
    } else {
        if ( p->p_master->in_running_q_count > 0 )
            ttds_queue_insert_tail( &p->p_master->list_suspended, &p->node );
        else {
            __jobq_add_job(p_this, p);
        }
    }
    ttds_unlock(lk);
#ifdef TTDS_USING_JOB_HOOK
    if (p_this->hook){
        p_this->hook( p_this->arg );
    }
#endif
}

void ttds_cjobq_poll_timer(struct ttds_cjobq *p_this, int tick_add)
{
    ttds_queue_t h,timeoutlist;
    struct ttds_cjob *j = 0;
    struct ttds_cjob *tmp_next = 0;
    int32_t next_timeout_tick = 0;
    int32_t tick_min = -1;
    int32_t diff = 0;
    ttds_lock_decl(lk);

    ttds_queue_init(&timeoutlist);
    ttds_lock(lk);
    if (ttds_queue_empty(&p_this->job_list_delayed)){
        p_this->tick_next_timeout = -1;
        ttds_unlock(lk);
        return ;
    }
    p_this->tick_next_timeout -= tick_add;
    if ( p_this->tick_next_timeout > 0){
        p_this->tick_pass += tick_add;
        ttds_unlock(lk);
        return ;
    }
    ttds_queue_move(&p_this->job_list_delayed, &h);
    tick_add += p_this->tick_pass;
    p_this->tick_pass = 0;
    ttds_unlock(lk);
    ttds_queue_for_each_container_safe(j, tmp_next,&h, struct ttds_cjob,node)
    {
        j->tick -= tick_add;
        if (j->tick <= 0)
        {
            ttds_queue_remove(&j->node);
            ttds_queue_insert_tail( &timeoutlist, &j->node );
        } else {
            if (tick_min >= 0){
                tick_min = ttds_min(tick_min, j->tick);
            } else {
                tick_min = j->tick;
            }
        }
    }
    ttds_lock(lk);
    ttds_queue_splice_tail(&p_this->job_list_delayed, &h );
    ttds_queue_splice_tail(&p_this->job_list, &timeoutlist );
    p_this->tick_next_timeout = tick_min;
    ttds_unlock(lk);
}

int ttds_cjobq_is_empty(struct ttds_cjobq *p_this )
{
    int ok = 0;
    ttds_lock_decl(lk);
    ttds_lock(lk);
    ok = ttds_queue_empty( &p_this->job_list );
    ttds_unlock(lk);

    return ok;
}
int ttds_cjobq_poll(struct ttds_cjobq *p_this, int tick_add, jmp_buf **jb_parent )
{
    ttds_queue_t h;
    jmp_buf *backup = 0;
    jmp_buf jb;
    ttds_lock_decl(lk);
    if (jb_parent)
    {
        backup = *jb_parent;
        *jb_parent = &jb;
    }
    ttds_queue_init(&h);

    if (tick_add)
        ttds_cjobq_poll_timer(p_this, tick_add);

    ttds_lock(lk);
    if (ttds_queue_empty(&p_this->job_list)){
        ttds_unlock(lk);
        if (jb_parent)
            *jb_parent = backup;
        return 0;
    }
    ttds_queue_move(&p_this->job_list, &h);
    ttds_queue_init(&p_this->job_list);
    ttds_unlock(lk);
    struct ttds_cjob *j = 0;
    struct ttds_cjob *tmp_next = 0;
    uint32_t has_exception = 0;

    ttds_queue_for_each_container_safe(j, tmp_next,&h, struct ttds_cjob,node)
    {
        uint32_t ret = 0;
        j->p_running_q = p_this;
        p_this->p_running_j = j;
        if (j->func) {
            if (j->p_master){
                j->p_master->in_running_q_count--;
            }
            if (jb_parent){
                ret = setjmp(**jb_parent);
            }
            if (ret == 0){
                j->func(j);
            } else {
                if (has_exception == 0)
                    has_exception = ret;
            }
            if (j->p_master){
                if (j->p_master->in_running_q_count == 0){
                    ttds_lock(lk);
                    if (!ttds_queue_empty(&j->p_master->list_suspended))
                    {
                        ttds_queue_t *h = &j->p_master->list_suspended;
                        ttds_queue_t *p = h->next;
                        ttds_queue_remove(p);
                        struct ttds_cjob* p_ptr = ttds_queue_container(p, struct ttds_cjob, node);
                        p_ptr->p_master->in_running_q_count++;
                        ttds_queue_insert_tail( &p_this->job_list, p );
                    }
                    ttds_unlock(lk);
                }
            }
        }
        p_this->p_running_j = 0;
        {
            int need_destroy = 0;
            ttds_lock(lk);
            j->ref--;
            if (j->ref > 0){
                ttds_queue_remove(&j->node);
            } else {
                need_destroy = 1;
            }
            ttds_unlock(lk);
            if (need_destroy)
                ttds_cjob_destroy(j);
        }
    }
    int has_job = 1;
    ttds_lock(lk);
    ttds_queue_splice_tail(&p_this->job_free, &h );
    if (ttds_queue_empty(&p_this->job_list))
    {
        has_job = 0;
    }
    ttds_unlock(lk);
    if (jb_parent)
    {
        *jb_parent = backup;
        if (has_exception){
            longjmp(**jb_parent, has_exception);
        }
    }
    return has_job;
}

